<!DOCTYPE html>
<html>

<head>
  <title>Quarkus - Using Apache Kafka Streams</title>
  <script id="adobe_dtm" src="https://www.redhat.com/dtm.js" type="text/javascript"></script>
  <script src="/assets/javascript/highlight.pack.js" type="text/javascript"></script>
  <META HTTP-EQUIV='Content-Security-Policy' CONTENT="default-src 'none'; script-src 'self' 'unsafe-eval' 'sha256-ANpuoVzuSex6VhqpYgsG25OHWVA1I+F6aGU04LoI+5s=' 'sha256-ipy9P/3rZZW06mTLAR0EnXvxSNcnfSDPLDuh3kzbB1w=' js.bizographics.com https://www.redhat.com assets.adobedtm.com jsonip.com https://ajax.googleapis.com https://www.googletagmanager.com https://www.google-analytics.com https://use.fontawesome.com; style-src 'self' https://fonts.googleapis.com https://use.fontawesome.com; img-src 'self' *; media-src 'self' ; frame-src https://www.googletagmanager.com https://www.youtube.com; frame-ancestors 'none'; base-uri 'none'; object-src 'none'; form-action 'none'; font-src 'self' https://use.fontawesome.com https://fonts.gstatic.com;">
  <META HTTP-EQUIV='X-Frame-Options' CONTENT="DENY">
  <META HTTP-EQUIV='X-XSS-Protection' CONTENT="1; mode=block">
  <META HTTP-EQUIV='X-Content-Type-Options' CONTENT="nosniff">
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <meta name="description" content="Quarkus: Supersonic Subatomic Java">
  <meta name="twitter:card" content="summary_large_image">
  <meta name="twitter:site" content="@QuarkusIO"> 
  <meta name="twitter:creator" content="@QuarkusIO">
  <meta property="og:url" content="https://quarkus.io/guides/kafka-streams" />
  <meta property="og:title" content="Quarkus - Using Apache Kafka Streams" />
  <meta property="og:description" content="Quarkus: Supersonic Subatomic Java" />
  <meta property="og:image" content="/assets/images/quarkus_card.png" />
  <link rel="canonical" href="https://quarkus.io/guides/kafka-streams">
  <link rel="shortcut icon" type="image/png" href="/favicon.ico" >
  <link rel="stylesheet" href="https://quarkus.io/guides/stylesheet/config.css" />
  <link rel="stylesheet" href="/assets/css/main.css" />
  <link rel="stylesheet" href="https://use.fontawesome.com/releases/v5.1.0/css/all.css" integrity="sha384-lKuwvrZot6UHsBSfcMvOkWwlCMgc0TaWr+30HWe3a4ltaBwTZhyTEggF5tJv8tbt" crossorigin="anonymous">
  <link rel="alternate" type="application/rss+xml"  href="https://quarkus.io/feed.xml" title="Quarkus">
  <script src="https://quarkus.io/assets/javascript/goan.js" type="text/javascript"></script>
  <script src="https://quarkus.io/assets/javascript/hl.js" type="text/javascript"></script>
</head>

<body class="guides">
  <!-- Google Tag Manager (noscript) -->
  <noscript><iframe src="https://www.googletagmanager.com/ns.html?id=GTM-NJWS5L"
  height="0" width="0" style="display:none;visibility:hidden"></iframe></noscript>
  <!-- End Google Tag Manager (noscript) -->

  <div class="nav-wrapper">
  <div class="grid-wrapper">
    <div class="width-12-12">
      <input type="checkbox" id="checkbox" />
      <nav id="main-nav" class="main-nav">
  <div class="container">
    <div class="logo-wrapper">
      
        <a href="/"><img src="/assets/images/quarkus_logo_horizontal_rgb_600px_reverse.png" class="project-logo" title="Quarkus"></a>
      
    </div>
    <label class="nav-toggle" for="checkbox">
      <i class="fa fa-bars"></i>
    </label>
    <div id="menu" class="menu">
      <span>
        <a href="/get-started/" class="">Get Started</a>
      </span>
      <span>
        <a href="/guides/" class="active">Guides</a>
      </span>
      <span>
        <a href="/community/" class="">Community</a>
      </span>
      <span>
        <a href="/support/" class="">Support</a>
      </span>
      <span>
        <a href="/blog/" class="">Blog</a>
      </span>
      <span>
        <a href="https://code.quarkus.io" class="button-cta secondary white">Start Coding</a>
      </span>
    </div>
  </div>
      </nav>
    </div>
  </div>
</div>

  <div class="content">
    <div class="guide">
  <div class="width-12-12">
    <h1 class="text-caps">Quarkus - Using Apache Kafka Streams</h1>
    <div class="hide-mobile toc"><ul class="sectlevel1">
<li><a href="#prerequisites">Prerequisites</a></li>
<li><a href="#architecture">Architecture</a></li>
<li><a href="#solution">Solution</a></li>
<li><a href="#creating-the-producer-maven-project">Creating the Producer Maven Project</a>
<ul class="sectlevel2">
<li><a href="#the-temperature-value-producer">The Temperature Value Producer</a></li>
<li><a href="#topic-configuration">Topic Configuration</a></li>
</ul>
</li>
<li><a href="#creating-the-aggregator-maven-project">Creating the Aggregator Maven Project</a>
<ul class="sectlevel2">
<li><a href="#the-pipeline-implementation">The Pipeline Implementation</a></li>
</ul>
</li>
<li><a href="#building-and-running-the-applications">Building and Running the Applications</a></li>
<li><a href="#interactive-queries">Interactive Queries</a></li>
<li><a href="#scaling-out">Scaling Out</a></li>
<li><a href="#running-natively">Running Natively</a></li>
<li><a href="#kafka-streams-health-checks">Kafka Streams Health Checks</a>
<ul class="sectlevel2">
<li><a href="#liveness-health-check">Liveness health check</a></li>
<li><a href="#readiness-health-check">Readiness health check</a></li>
</ul>
</li>
<li><a href="#going-further">Going Further</a></li>
<li><a href="#configuration-reference">Configuration Reference</a></li>
</ul></div>
    <div>
      <div id="preamble">
<div class="sectionbody">
<div class="paragraph">
<p>This guide demonstrates how your Quarkus application can utilize the Apache Kafka Streams API to implement stream processing applications based on Apache Kafka.</p>
</div>
</div>
</div>
<div class="sect1">
<h2 id="prerequisites"><a class="anchor" href="#prerequisites"></a>Prerequisites</h2>
<div class="sectionbody">
<div class="paragraph">
<p>To complete this guide, you need:</p>
</div>
<div class="ulist">
<ul>
<li>
<p>less than 30 minutes</p>
</li>
<li>
<p>an IDE</p>
</li>
<li>
<p>JDK 1.8+ installed with <code>JAVA_HOME</code> configured appropriately</p>
</li>
<li>
<p>Apache Maven 3.6.2+</p>
</li>
<li>
<p>Docker Compose to start an Apache Kafka development cluster</p>
</li>
<li>
<p>GraalVM installed if you want to run in native mode.</p>
</li>
</ul>
</div>
<div class="paragraph">
<p>It is recommended, that you have read the <a href="https://github.com/quarkusio/quarkus-quickstarts/tree/master/kafka-quickstart">Kafka quickstart</a> before.</p>
</div>
<div class="admonitionblock note">
<table>
<tr>
<td class="icon">
<i class="fa icon-note" title="Note"></i>
</td>
<td class="content">
<div class="paragraph">
<p>The Quarkus extension for Kafka Streams allows for very fast turnaround times during development by supporting the Quarkus Dev Mode (e.g. via <code>./mvnw compile quarkus:dev</code>).
After changing the code of your Kafka Streams topology, the application will automatically be reloaded when the next input message arrives.</p>
</div>
<div class="paragraph">
<p>A recommended development set-up is to have some producer which creates test messages on the processed topic(s) in fixed intervals, e.g. every second and observe the streaming application&#8217;s output topic(s) using a tool such as <code>kafkacat</code>.
Using the dev mode, you&#8217;ll instantly see messages on the output topic(s) as produced by the latest version of your streaming application when saving.</p>
</div>
<div class="paragraph">
<p>For the best development experience, we recommend applying the following configuration settings to your Kafka broker:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs">group.min.session.timeout.ms=250</code></pre>
</div>
</div>
<div class="paragraph">
<p>Also specify the following settings in your Quarkus <code>application.properties</code>:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs">kafka-streams.consumer.session.timeout.ms=250
kafka-streams.consumer.heartbeat.interval.ms=200</code></pre>
</div>
</div>
<div class="paragraph">
<p>Together, these settings will ensure that the application can very quickly reconnect to the broker after being restarted in dev mode.</p>
</div>
</td>
</tr>
</table>
</div>
</div>
</div>
<div class="sect1">
<h2 id="architecture"><a class="anchor" href="#architecture"></a>Architecture</h2>
<div class="sectionbody">
<div class="paragraph">
<p>In this guide, we are going to generate (random) temperature values in one component (named <code>generator</code>).
These values are associated to given weather stations and are written in a Kafka topic (<code>temperature-values</code>).
Another topic (<code>weather-stations</code>) contains just the master data about the weather stations themselves (id and name).</p>
</div>
<div class="paragraph">
<p>A second component (<code>aggregator</code>) reads from the two Kafka topics and processes them in a streaming pipeline:</p>
</div>
<div class="ulist">
<ul>
<li>
<p>the two topics are joined on weather station id</p>
</li>
<li>
<p>per weather station the min, max and average temperature is determined</p>
</li>
<li>
<p>this aggregated data is written out to a third topic (<code>temperatures-aggregated</code>)</p>
</li>
</ul>
</div>
<div class="paragraph">
<p>The data can be examined by inspecting the output topic.
By exposing a Kafka Streams <a href="https://kafka.apache.org/22/documentation/streams/developer-guide/interactive-queries.html">interactive query</a>,
the latest result for each weather station can alternatively be obtained via a simple REST query.</p>
</div>
<div class="paragraph">
<p>The overall architecture looks like so:</p>
</div>
<div class="imageblock">
<div class="content">
<img src="/guides/images/kafka-streams-guide-architecture.png" alt="Architecture">
</div>
</div>
</div>
</div>
<div class="sect1">
<h2 id="solution"><a class="anchor" href="#solution"></a>Solution</h2>
<div class="sectionbody">
<div class="paragraph">
<p>We recommend that you follow the instructions in the next sections and create the application step by step.
However, you can go right to the completed example.</p>
</div>
<div class="paragraph">
<p>Clone the Git repository: <code>git clone <a href="https://github.com/quarkusio/quarkus-quickstarts.git" class="bare">https://github.com/quarkusio/quarkus-quickstarts.git</a></code>, or download an <a href="https://github.com/quarkusio/quarkus-quickstarts/archive/master.zip">archive</a>.</p>
</div>
<div class="paragraph">
<p>The solution is located in the <code>kafka-streams-quickstart</code> <a href="https://github.com/quarkusio/quarkus-quickstarts/tree/master/kafka-streams-quickstart">directory</a>.</p>
</div>
</div>
</div>
<div class="sect1">
<h2 id="creating-the-producer-maven-project"><a class="anchor" href="#creating-the-producer-maven-project"></a>Creating the Producer Maven Project</h2>
<div class="sectionbody">
<div class="paragraph">
<p>First, we need a new project with the temperature value producer.
Create a new project with the following command:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs">mvn io.quarkus:quarkus-maven-plugin:1.7.0.Final:create \
    -DprojectGroupId=org.acme \
    -DprojectArtifactId=kafka-streams-quickstart-producer \
    -Dextensions="kafka" \
    &amp;&amp; mv kafka-streams-quickstart-producer producer</code></pre>
</div>
</div>
<div class="paragraph">
<p>This command generates a Maven project, importing the Reactive Messaging and Kafka connector extensions.</p>
</div>
<div class="paragraph">
<p>If you already have your Quarkus project configured, you can add the <code>smallrye-reactive-messaging-kafka</code> extension
to your project by running the following command in your project base directory:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="bash" class="language-bash hljs">./mvnw quarkus:add-extension -Dextensions="kafka"</code></pre>
</div>
</div>
<div class="paragraph">
<p>This will add the following to your <code>pom.xml</code>:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="xml" class="language-xml hljs">&lt;dependency&gt;
    &lt;groupId&gt;io.quarkus&lt;/groupId&gt;
    &lt;artifactId&gt;quarkus-smallrye-reactive-messaging-kafka&lt;/artifactId&gt;
&lt;/dependency&gt;</code></pre>
</div>
</div>
<div class="sect2">
<h3 id="the-temperature-value-producer"><a class="anchor" href="#the-temperature-value-producer"></a>The Temperature Value Producer</h3>
<div class="paragraph">
<p>Create the <code>producer/src/main/java/org/acme/kafka/streams/producer/generator/ValuesGenerator.java</code> file,
with the following content:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="java" class="language-java hljs">package org.acme.kafka.streams.producer.generator;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.Instant;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import javax.enterprise.context.ApplicationScoped;

import org.eclipse.microprofile.reactive.messaging.Outgoing;
import org.jboss.logging.Logger;

import io.reactivex.Flowable;
import io.smallrye.reactive.messaging.kafka.KafkaRecord;

/**
 * A bean producing random temperature data every second.
 * The values are written to a Kafka topic (temperature-values).
 * Another topic contains the name of weather stations (weather-stations).
 * The Kafka configuration is specified in the application configuration.
 */
@ApplicationScoped
public class ValuesGenerator {

    private static final Logger LOG = Logger.getLogger(ValuesGenerator.class);

    private Random random = new Random();

    private List&lt;WeatherStation&gt; stations = Collections.unmodifiableList(
            Arrays.asList(
                    new WeatherStation(1, "Hamburg", 13),
                    new WeatherStation(2, "Snowdonia", 5),
                    new WeatherStation(3, "Boston", 11),
                    new WeatherStation(4, "Tokio", 16),
                    new WeatherStation(5, "Cusco", 12),
                    new WeatherStation(6, "Svalbard", -7),
                    new WeatherStation(7, "Porthsmouth", 11),
                    new WeatherStation(8, "Oslo", 7),
                    new WeatherStation(9, "Marrakesh", 20)
            ));


    @Outgoing("temperature-values")                             <i class="conum" data-value="1"></i><b>(1)</b>
    public Flowable&lt;KafkaRecord&lt;Integer, String&gt;&gt; generate() {

        return Flowable.interval(500, TimeUnit.MILLISECONDS)    <i class="conum" data-value="2"></i><b>(2)</b>
                .onBackpressureDrop()
                .map(tick -&gt; {
                    WeatherStation station = stations.get(random.nextInt(stations.size()));
                    double temperature = BigDecimal.valueOf(random.nextGaussian() * 15 + station.averageTemperature)
                            .setScale(1, RoundingMode.HALF_UP)
                            .doubleValue();

                    LOG.infov("station: {0}, temperature: {1}", station.name, temperature);
                    return KafkaRecord.of(station.id, Instant.now() + ";" + temperature);
                });
    }

    @Outgoing("weather-stations")                               <i class="conum" data-value="3"></i><b>(3)</b>
    public Flowable&lt;KafkaRecord&lt;Integer, String&gt;&gt; weatherStations() {
        List&lt;KafkaRecord&lt;Integer, String&gt;&gt; stationsAsJson = stations.stream()
            .map(s -&gt; KafkaRecord.of(
                    s.id,
                    "{ \"id\" : " + s.id +
                    ", \"name\" : \"" + s.name + "\" }"))
            .collect(Collectors.toList());

        return Flowable.fromIterable(stationsAsJson);
    };

    private static class WeatherStation {

        int id;
        String name;
        int averageTemperature;

        public WeatherStation(int id, String name, int averageTemperature) {
            this.id = id;
            this.name = name;
            this.averageTemperature = averageTemperature;
        }
    }
}</code></pre>
</div>
</div>
<div class="colist arabic">
<table>
<tr>
<td><i class="conum" data-value="1"></i><b>1</b></td>
<td>Instruct Reactive Messaging to dispatch the items from the returned <code>Flowable</code> to <code>temperature-values</code>.</td>
</tr>
<tr>
<td><i class="conum" data-value="2"></i><b>2</b></td>
<td>The method returns a RX Java 2 <em>stream</em> (<code>Flowable</code>) emitting a random temperature value every 0.5 seconds.</td>
</tr>
<tr>
<td><i class="conum" data-value="3"></i><b>3</b></td>
<td>Instruct Reactive Messaging to dispatch the items from the returned <code>Flowable</code> (static list of weather stations) to <code>weather-stations</code>.</td>
</tr>
</table>
</div>
<div class="paragraph">
<p>The two methods each return a <em>reactive stream</em> whose items are sent to the streams named <code>temperature-values</code> and <code>weather-stations</code>, respectively.</p>
</div>
</div>
<div class="sect2">
<h3 id="topic-configuration"><a class="anchor" href="#topic-configuration"></a>Topic Configuration</h3>
<div class="paragraph">
<p>The two channels are mapped to Kafka topics using the Quarkus configuration file <code>application.properties</code>.
For that, add the following to the file <code>producer/src/main/resources/application.properties</code>:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs"># Configure the Kafka broker location
kafka.bootstrap.servers=localhost:9092

mp.messaging.outgoing.temperature-values.connector=smallrye-kafka
mp.messaging.outgoing.temperature-values.key.serializer=org.apache.kafka.common.serialization.IntegerSerializer
mp.messaging.outgoing.temperature-values.value.serializer=org.apache.kafka.common.serialization.StringSerializer

mp.messaging.outgoing.weather-stations.connector=smallrye-kafka
mp.messaging.outgoing.weather-stations.key.serializer=org.apache.kafka.common.serialization.IntegerSerializer
mp.messaging.outgoing.weather-stations.value.serializer=org.apache.kafka.common.serialization.StringSerializer</code></pre>
</div>
</div>
<div class="paragraph">
<p>This configures the Kafka bootstrap server, the two topics and the corresponding (de-)serializers.
More details about the different configuration options are available on the <a href="https://kafka.apache.org/documentation/#producerconfigs">Producer configuration</a> and <a href="https://kafka.apache.org/documentation/#consumerconfigs">Consumer configuration</a> section from the Kafka documentation.</p>
</div>
</div>
</div>
</div>
<div class="sect1">
<h2 id="creating-the-aggregator-maven-project"><a class="anchor" href="#creating-the-aggregator-maven-project"></a>Creating the Aggregator Maven Project</h2>
<div class="sectionbody">
<div class="paragraph">
<p>With the producer application in place, it&#8217;s time to implement the actual aggregator application,
which will run the Kafka Streams pipeline.
Create another project like so:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs">mvn io.quarkus:quarkus-maven-plugin:1.7.0.Final:create \
    -DprojectGroupId=org.acme \
    -DprojectArtifactId=kafka-streams-quickstart-aggregator \
    -Dextensions="kafka-streams,resteasy-jsonb" \
    &amp;&amp; mv kafka-streams-quickstart-aggregator aggregator</code></pre>
</div>
</div>
<div class="paragraph">
<p>This creates the <code>aggregator</code> project with the Quarkus extension for Kafka Streams and with RESTEasy support for JSON-B.</p>
</div>
<div class="paragraph">
<p>If you already have your Quarkus project configured, you can add the <code>kafka-streams</code> extension
to your project by running the following command in your project base directory:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="bash" class="language-bash hljs">./mvnw quarkus:add-extension -Dextensions="kafka-streams"</code></pre>
</div>
</div>
<div class="paragraph">
<p>This will add the following to your <code>pom.xml</code>:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="xml" class="language-xml hljs">&lt;dependency&gt;
    &lt;groupId&gt;io.quarkus&lt;/groupId&gt;
    &lt;artifactId&gt;quarkus-kafka-streams&lt;/artifactId&gt;
&lt;/dependency&gt;</code></pre>
</div>
</div>
<div class="sect2">
<h3 id="the-pipeline-implementation"><a class="anchor" href="#the-pipeline-implementation"></a>The Pipeline Implementation</h3>
<div class="paragraph">
<p>Let&#8217;s begin the implementation of the stream processing application by creating
a few value objects for representing temperature measurements, weather stations and for keeping track of aggregated values.</p>
</div>
<div class="paragraph">
<p>First, create the file <code>aggregator/src/main/java/org/acme/kafka/streams/aggregator/model/WeatherStation.java</code>,
representing a weather station, with the following content:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="java" class="language-java hljs">package org.acme.kafka.streams.aggregator.model;

import io.quarkus.runtime.annotations.RegisterForReflection;

@RegisterForReflection <i class="conum" data-value="1"></i><b>(1)</b>
public class WeatherStation {

    public int id;
    public String name;
}</code></pre>
</div>
</div>
<div class="colist arabic">
<table>
<tr>
<td><i class="conum" data-value="1"></i><b>1</b></td>
<td>By adding the <code>@RegisterForReflection</code> annotation, it is ensured that this type can be instantiated reflectively when running the application in native mode.</td>
</tr>
</table>
</div>
<div class="paragraph">
<p>Then the file <code>aggregator/src/main/java/org/acme/kafka/streams/aggregator/model/TemperatureMeasurement.java</code>,
representing temperature measurements for a given station:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="java" class="language-java hljs">package org.acme.kafka.streams.aggregator.model;

import java.time.Instant;

public class TemperatureMeasurement {

    public int stationId;
    public String stationName;
    public Instant timestamp;
    public double value;

    public TemperatureMeasurement(int stationId, String stationName, Instant timestamp,
            double value) {
        this.stationId = stationId;
        this.stationName = stationName;
        this.timestamp = timestamp;
        this.value = value;
    }
}</code></pre>
</div>
</div>
<div class="paragraph">
<p>And finally <code>aggregator/src/main/java/org/acme/kafka/streams/aggregator/model/Aggregation.java</code>,
which will be used to keep track of the aggregated values while the events are processed in the streaming pipeline:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="java" class="language-java hljs">package org.acme.kafka.streams.aggregator.model;

import java.math.BigDecimal;
import java.math.RoundingMode;

import io.quarkus.runtime.annotations.RegisterForReflection;

@RegisterForReflection
public class Aggregation {

    public int stationId;
    public String stationName;
    public double min = Double.MAX_VALUE;
    public double max = Double.MIN_VALUE;
    public int count;
    public double sum;
    public double avg;

    public Aggregation updateFrom(TemperatureMeasurement measurement) {
        stationId = measurement.stationId;
        stationName = measurement.stationName;

        count++;
        sum += measurement.value;
        avg = BigDecimal.valueOf(sum / count)
                .setScale(1, RoundingMode.HALF_UP).doubleValue();

        min = Math.min(min, measurement.value);
        max = Math.max(max, measurement.value);

        return this;
    }
}</code></pre>
</div>
</div>
<div class="paragraph">
<p>Next, let&#8217;s create the actual streaming query implementation itself in the <code>aggregator/src/main/java/org/acme/kafka/streams/aggregator/streams/TopologyProducer.java</code> file.
All we need to do for that is to declare a CDI producer method which returns the Kafka Streams <code>Topology</code>;
the Quarkus extension will take care of configuring, starting and stopping the actual Kafka Streams engine.</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="java" class="language-java hljs">package org.acme.kafka.streams.aggregator.streams;

import java.time.Instant;

import javax.enterprise.context.ApplicationScoped;
import javax.enterprise.inject.Produces;

import org.acme.kafka.streams.aggregator.model.Aggregation;
import org.acme.kafka.streams.aggregator.model.TemperatureMeasurement;
import org.acme.kafka.streams.aggregator.model.WeatherStation;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.GlobalKTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
import org.apache.kafka.streams.state.Stores;

import io.quarkus.kafka.client.serialization.JsonbSerde;

@ApplicationScoped
public class TopologyProducer {

    static final String WEATHER_STATIONS_STORE = "weather-stations-store";

    private static final String WEATHER_STATIONS_TOPIC = "weather-stations";
    private static final String TEMPERATURE_VALUES_TOPIC = "temperature-values";
    private static final String TEMPERATURES_AGGREGATED_TOPIC = "temperatures-aggregated";

    @Produces
    public Topology buildTopology() {
        StreamsBuilder builder = new StreamsBuilder();

        JsonbSerde&lt;WeatherStation&gt; weatherStationSerde = new JsonbSerde&lt;&gt;(
                WeatherStation.class);
        JsonbSerde&lt;Aggregation&gt; aggregationSerde = new JsonbSerde&lt;&gt;(Aggregation.class);

        KeyValueBytesStoreSupplier storeSupplier = Stores.persistentKeyValueStore(
                WEATHER_STATIONS_STORE);

        GlobalKTable&lt;Integer, WeatherStation&gt; stations = builder.globalTable( <i class="conum" data-value="1"></i><b>(1)</b>
                WEATHER_STATIONS_TOPIC,
                Consumed.with(Serdes.Integer(), weatherStationSerde));

        builder.stream(                                                       <i class="conum" data-value="2"></i><b>(2)</b>
                        TEMPERATURE_VALUES_TOPIC,
                        Consumed.with(Serdes.Integer(), Serdes.String())
                )
                .join(                                                        <i class="conum" data-value="3"></i><b>(3)</b>
                        stations,
                        (stationId, timestampAndValue) -&gt; stationId,
                        (timestampAndValue, station) -&gt; {
                            String[] parts = timestampAndValue.split(";");
                            return new TemperatureMeasurement(station.id, station.name,
                                    Instant.parse(parts[0]), Double.valueOf(parts[1]));
                        }
                )
                .groupByKey()                                                 <i class="conum" data-value="4"></i><b>(4)</b>
                .aggregate(                                                   <i class="conum" data-value="5"></i><b>(5)</b>
                        Aggregation::new,
                        (stationId, value, aggregation) -&gt; aggregation.updateFrom(value),
                        Materialized.&lt;Integer, Aggregation&gt; as(storeSupplier)
                            .withKeySerde(Serdes.Integer())
                            .withValueSerde(aggregationSerde)
                )
                .toStream()
                .to(                                                          <i class="conum" data-value="6"></i><b>(6)</b>
                        TEMPERATURES_AGGREGATED_TOPIC,
                        Produced.with(Serdes.Integer(), aggregationSerde)
                );

        return builder.build();
    }
}</code></pre>
</div>
</div>
<div class="colist arabic">
<table>
<tr>
<td><i class="conum" data-value="1"></i><b>1</b></td>
<td>The <code>weather-stations</code> table is read into a <code>GlobalKTable</code>, representing the current state of each weather station</td>
</tr>
<tr>
<td><i class="conum" data-value="2"></i><b>2</b></td>
<td>The <code>temperature-values</code> topic is read into a <code>KStream</code>; whenever a new message arrives to this topic, the pipeline will be processed for this measurement</td>
</tr>
<tr>
<td><i class="conum" data-value="3"></i><b>3</b></td>
<td>The message from the <code>temperature-values</code> topic is joined with the corresponding weather station, using the topic&#8217;s key (weather station id); the join result contains the data from the measurement and associated weather station message</td>
</tr>
<tr>
<td><i class="conum" data-value="4"></i><b>4</b></td>
<td>The values are grouped by message key (the weather station id)</td>
</tr>
<tr>
<td><i class="conum" data-value="5"></i><b>5</b></td>
<td>Within each group, all the measurements of that station are aggregated, by keeping track of minimum and maximum values and calculating the average value of all measurements of that station (see the <code>Aggregation</code> type)</td>
</tr>
<tr>
<td><i class="conum" data-value="6"></i><b>6</b></td>
<td>The results of the pipeline are written out to the <code>temperatures-aggregated</code> topic</td>
</tr>
</table>
</div>
<div class="paragraph">
<p>The Kafka Streams extension is configured via the Quarkus configuration file <code>application.properties</code>.
Create the file <code>aggregator/src/main/resources/application.properties</code> with the following contents:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs">quarkus.kafka-streams.bootstrap-servers=localhost:9092
quarkus.kafka-streams.application-server=${hostname}:8080
quarkus.kafka-streams.topics=weather-stations,temperature-values

# pass-through options
kafka-streams.cache.max.bytes.buffering=10240
kafka-streams.commit.interval.ms=1000
kafka-streams.metadata.max.age.ms=500
kafka-streams.auto.offset.reset=earliest
kafka-streams.metrics.recording.level=DEBUG</code></pre>
</div>
</div>
<div class="paragraph">
<p>The options with the <code>quarkus.kafka-streams</code> prefix can be changed dynamically at application startup,
e.g. via environment variables or system properties.
<code>bootstrap-servers</code> and <code>application-server</code> are mapped to the Kafka Streams properties <code>bootstrap.servers</code> and <code>application.server</code>, respectively.
<code>topics</code> is specific to Quarkus: the application will wait for all the given topics to exist before launching the Kafka Streams engine.
This is to done to gracefully await the creation of topics that don&#8217;t yet exist at application startup time.</p>
</div>
<div class="paragraph">
<p>All the properties within the <code>kafka-streams</code> namespace are passed through as-is to the Kafka Streams engine.
Changing their values requires a rebuild of the application.</p>
</div>
</div>
</div>
</div>
<div class="sect1">
<h2 id="building-and-running-the-applications"><a class="anchor" href="#building-and-running-the-applications"></a>Building and Running the Applications</h2>
<div class="sectionbody">
<div class="paragraph">
<p>We now can build the <code>producer</code> and <code>aggregator</code> applications:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs">./mvnw clean package -f producer/pom.xml
./mvnw clean package -f aggregator/pom.xml</code></pre>
</div>
</div>
<div class="paragraph">
<p>Instead of running them directly on the host machine using the Quarkus dev mode,
we&#8217;re going to package them into container images and launch them via Docker Compose.
This is done in order to demonstrate scaling the <code>aggregator</code> aggregation to multiple nodes later on.</p>
</div>
<div class="paragraph">
<p>The <code>Dockerfile</code> created by Quarkus by default needs one adjustment for the <code>aggregator</code> application in order to run the Kafka Streams pipeline.
To do so, edit the file <code>aggregator/src/main/docker/Dockerfile.jvm</code> and replace the line <code>FROM fabric8/java-alpine-openjdk8-jre</code> with <code>FROM fabric8/java-centos-openjdk8-jdk</code>.</p>
</div>
<div class="paragraph">
<p>Next create a Docker Compose file (<code>docker-compose.yaml</code>) for spinning up the two applications as well as Apache Kafka and ZooKeeper like so:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="yaml" class="language-yaml hljs">version: '3.5'

services:
  zookeeper:
    image: strimzi/kafka:0.11.3-kafka-2.1.0
    command: [
      "sh", "-c",
      "bin/zookeeper-server-start.sh config/zookeeper.properties"
    ]
    ports:
      - "2181:2181"
    environment:
      LOG_DIR: /tmp/logs
    networks:
      - kafkastreams-network
  kafka:
    image: strimzi/kafka:0.11.3-kafka-2.1.0
    command: [
      "sh", "-c",
      "bin/kafka-server-start.sh config/server.properties --override listeners=$${KAFKA_LISTENERS} --override advertised.listeners=$${KAFKA_ADVERTISED_LISTENERS} --override zookeeper.connect=$${KAFKA_ZOOKEEPER_CONNECT} --override num.partitions=$${KAFKA_NUM_PARTITIONS}"
    ]
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      LOG_DIR: "/tmp/logs"
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_NUM_PARTITIONS: 3
    networks:
      - kafkastreams-network

  producer:
    image: quarkus-quickstarts/kafka-streams-producer:1.0
    build:
      context: producer
      dockerfile: src/main/docker/Dockerfile.${QUARKUS_MODE:-jvm}
    environment:
      KAFKA_BOOTSTRAP_SERVERS: kafka:9092
    networks:
      - kafkastreams-network

  aggregator:
    image: quarkus-quickstarts/kafka-streams-aggregator:1.0
    build:
      context: aggregator
      dockerfile: src/main/docker/Dockerfile.${QUARKUS_MODE:-jvm}
    environment:
      QUARKUS_KAFKA_STREAMS_BOOTSTRAP_SERVERS: kafka:9092
    networks:
      - kafkastreams-network

networks:
  kafkastreams-network:
    name: ks</code></pre>
</div>
</div>
<div class="paragraph">
<p>To launch all the containers, building the <code>producer</code> and <code>aggregator</code> container images,
run <code>docker-compose up --build</code>.</p>
</div>
<div class="paragraph">
<p>You should see log statements from the <code>producer</code> application about messages being sent to the "temperature-values" topic.</p>
</div>
<div class="paragraph">
<p>Now run an instance of the <em>debezium/tooling</em> image, attaching to the same network all the other containers run in.
This image provides several useful tools such as <em>kafkacat</em> and <em>httpie</em>:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs">docker run --tty --rm -i --network ks debezium/tooling:1.0</code></pre>
</div>
</div>
<div class="paragraph">
<p>Within the tooling container, run <em>kafkacat</em> to examine the results of the streaming pipeline:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs">kafkacat -b kafka:9092 -C -o beginning -q -t temperatures-aggregated

{"avg":34.7,"count":4,"max":49.4,"min":16.8,"stationId":9,"stationName":"Marrakesh","sum":138.8}
{"avg":15.7,"count":1,"max":15.7,"min":15.7,"stationId":2,"stationName":"Snowdonia","sum":15.7}
{"avg":12.8,"count":7,"max":25.5,"min":-13.8,"stationId":7,"stationName":"Porthsmouth","sum":89.7}
...</code></pre>
</div>
</div>
<div class="paragraph">
<p>You should see new values arrive as the producer continues to emit temperature measurements,
each value on the outbound topic showing the minimum, maximum and average temperature values of the represented weather station.</p>
</div>
</div>
</div>
<div class="sect1">
<h2 id="interactive-queries"><a class="anchor" href="#interactive-queries"></a>Interactive Queries</h2>
<div class="sectionbody">
<div class="paragraph">
<p>Subscribing to the <code>temperatures-aggregated</code> topic is a great way to react to any new temperature values.
It&#8217;s a bit wasteful though if you&#8217;re just interested in the latest aggregated value for a given weather station.
This is where Kafka Streams interactive queries shine:
they let you directly query the underlying state store of the pipeline for the value associated to a given key.
By exposing a simple REST endpoint which queries the state store,
the latest aggregation result can be retrieved without having to subscribe to any Kafka topic.</p>
</div>
<div class="paragraph">
<p>Let&#8217;s begin by creating a new class <code>InteractiveQueries</code> in the file <code>aggregator/src/main/java/org/acme/kafka/streams/aggregator/streams/InteractiveQueries.java</code>:</p>
</div>
<div class="paragraph">
<p>one more method to the <code>KafkaStreamsPipeline</code> class which obtains the current state for a given key:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="java" class="language-java hljs">package org.acme.kafka.streams.aggregator.streams;

import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;

import org.acme.kafka.streams.aggregator.model.Aggregation;
import org.acme.kafka.streams.aggregator.model.WeatherStationData;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.errors.InvalidStateStoreException;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;

@ApplicationScoped
public class InteractiveQueries {

    @Inject
    KafkaStreams streams;

    public GetWeatherStationDataResult getWeatherStationData(int id) {
        Aggregation result = getWeatherStationStore().get(id);

        if (result != null) {
            return GetWeatherStationDataResult.found(WeatherStationData.from(result)); <i class="conum" data-value="1"></i><b>(1)</b>
        }
        else {
            return GetWeatherStationDataResult.notFound();                             <i class="conum" data-value="2"></i><b>(2)</b>
        }
    }

    private ReadOnlyKeyValueStore&lt;Integer, Aggregation&gt; getWeatherStationStore() {
        while (true) {
            try {
                return streams.store(TopologyProducer.WEATHER_STATIONS_STORE, QueryableStoreTypes.keyValueStore());
            } catch (InvalidStateStoreException e) {
                // ignore, store not ready yet
            }
        }
    }
}</code></pre>
</div>
</div>
<div class="colist arabic">
<table>
<tr>
<td><i class="conum" data-value="1"></i><b>1</b></td>
<td>A value for the given station id was found, so that value will be returned</td>
</tr>
<tr>
<td><i class="conum" data-value="2"></i><b>2</b></td>
<td>No value was found, either because a non-existing station was queried or no measurement exists yet for the given station</td>
</tr>
</table>
</div>
<div class="paragraph">
<p>Also create the method&#8217;s return type in the file <code>aggregator/src/main/java/org/acme/kafka/streams/aggregator/streams/GetWeatherStationDataResult.java</code>:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="java" class="language-java hljs">package org.acme.kafka.streams.aggregator.streams;

import java.util.Optional;
import java.util.OptionalInt;

import org.acme.kafka.streams.aggregator.model.WeatherStationData;

public class GetWeatherStationDataResult {

    private static GetWeatherStationDataResult NOT_FOUND =
            new GetWeatherStationDataResult(null);

    private final WeatherStationData result;

    private GetWeatherStationDataResult(WeatherStationData result) {
        this.result = result;
    }

    public static GetWeatherStationDataResult found(WeatherStationData data) {
        return new GetWeatherStationDataResult(data);
    }

    public static GetWeatherStationDataResult notFound() {
        return NOT_FOUND;
    }

    public Optional&lt;WeatherStationData&gt; getResult() {
        return Optional.ofNullable(result);
    }
}</code></pre>
</div>
</div>
<div class="paragraph">
<p>Also create <code>aggregator/src/main/java/org/acme/kafka/streams/aggregator/model/WeatherStationData.java</code>,
which represents the actual aggregation result for a weather station:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="java" class="language-java hljs">package org.acme.kafka.streams.aggregator.model;

import io.quarkus.runtime.annotations.RegisterForReflection;

@RegisterForReflection
public class WeatherStationData {

    public int stationId;
    public String stationName;
    public double min = Double.MAX_VALUE;
    public double max = Double.MIN_VALUE;
    public int count;
    public double avg;

    private WeatherStationData(int stationId, String stationName, double min, double max,
            int count, double avg) {
        this.stationId = stationId;
        this.stationName = stationName;
        this.min = min;
        this.max = max;
        this.count = count;
        this.avg = avg;
    }

    public static WeatherStationData from(Aggregation aggregation) {
        return new WeatherStationData(
                aggregation.stationId,
                aggregation.stationName,
                aggregation.min,
                aggregation.max,
                aggregation.count,
                aggregation.avg);
    }
}</code></pre>
</div>
</div>
<div class="paragraph">
<p>We now can add a simple REST endpoint (<code>aggregator/src/main/java/org/acme/kafka/streams/aggregator/rest/WeatherStationEndpoint.java</code>),
which invokes <code>getWeatherStationData()</code> and returns the data to the client:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="java" class="language-java hljs">package org.acme.kafka.streams.aggregator.rest;

import java.net.URI;
import java.net.URISyntaxException;
import java.util.List;

import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
import javax.ws.rs.Consumes;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;

import org.acme.kafka.streams.aggregator.streams.GetWeatherStationDataResult;
import org.acme.kafka.streams.aggregator.streams.KafkaStreamsPipeline;

@ApplicationScoped
@Path("/weather-stations")
public class WeatherStationEndpoint {

    @Inject
    InteractiveQueries interactiveQueries;

    @GET
    @Path("/data/{id}")
    @Consumes(MediaType.APPLICATION_JSON)
    @Produces(MediaType.APPLICATION_JSON)
    public Response getWeatherStationData(@PathParam("id") int id) {
        GetWeatherStationDataResult result = interactiveQueries.getWeatherStationData(id);

        if (result.getResult().isPresent()) {  <i class="conum" data-value="1"></i><b>(1)</b>
            return Response.ok(result.getResult().get()).build();
        }
        else {
            return Response.status(Status.NOT_FOUND.getStatusCode(),
                    "No data found for weather station " + id).build();
        }
    }
}</code></pre>
</div>
</div>
<div class="colist arabic">
<table>
<tr>
<td><i class="conum" data-value="1"></i><b>1</b></td>
<td>Depending on whether a value was obtained, either return that value or a 404 response</td>
</tr>
</table>
</div>
<div class="paragraph">
<p>With this code in place, it&#8217;s time to rebuild the application and the <code>aggregator</code> service in Docker Compose:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs">./mvnw clean package -f aggregator/pom.xml
docker-compose stop aggregator
docker-compose up --build -d</code></pre>
</div>
</div>
<div class="paragraph">
<p>This will rebuild the <code>aggregator</code> container and restart its service.
Once that&#8217;s done, you can invoke the service&#8217;s REST API to obtain the temperature data for one of the existing stations.
To do so, you can use <code>httpie</code> in the tooling container launched before:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs">http aggregator:8080/weather-stations/data/1

HTTP/1.1 200 OK
Connection: keep-alive
Content-Length: 85
Content-Type: application/json
Date: Tue, 18 Jun 2019 19:29:16 GMT

{
    "avg": 12.9,
    "count": 146,
    "max": 41.0,
    "min": -25.6,
    "stationId": 1,
    "stationName": "Hamburg"
}</code></pre>
</div>
</div>
</div>
</div>
<div class="sect1">
<h2 id="scaling-out"><a class="anchor" href="#scaling-out"></a>Scaling Out</h2>
<div class="sectionbody">
<div class="paragraph">
<p>A very interesting trait of Kafka Streams applications is that they can be scaled out,
i.e. the load and state can be distributed amongst multiple application instances running the same pipeline.
Each node will then contain a subset of the aggregation results,
but Kafka Streams provides you with <a href="https://kafka.apache.org/22/documentation/streams/developer-guide/interactive-queries.html#querying-remote-state-stores-for-the-entire-app">an API</a> to obtain the information which node is hosting a given key.
The application can then either fetch the data directly from the other instance, or simply point the client to the location of that other node.</p>
</div>
<div class="paragraph">
<p>Launching multiple instances of the <code>aggregator</code> application will make look the overall architecture like so:</p>
</div>
<div class="imageblock">
<div class="content">
<img src="/guides/images/kafka-streams-guide-architecture-distributed.png" alt="Architecture with multiple aggregator nodes">
</div>
</div>
<div class="paragraph">
<p>The <code>InteractiveQueries</code> class must be adjusted slightly for this distributed architecture:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="java" class="language-java hljs">public GetWeatherStationDataResult getWeatherStationData(int id) {
    StreamsMetadata metadata = streams.metadataForKey(                  <i class="conum" data-value="1"></i><b>(1)</b>
            TopologyProducer.WEATHER_STATIONS_STORE,
            id,
            Serdes.Integer().serializer()
    );

    if (metadata == null || metadata == StreamsMetadata.NOT_AVAILABLE) {
        LOG.warn("Found no metadata for key {}", id);
        return GetWeatherStationDataResult.notFound();
    }
    else if (metadata.host().equals(host)) {                            <i class="conum" data-value="2"></i><b>(2)</b>
        LOG.info("Found data for key {} locally", id);
        Aggregation result = getWeatherStationStore().get(id);

        if (result != null) {
            return GetWeatherStationDataResult.found(WeatherStationData.from(result));
        }
        else {
            return GetWeatherStationDataResult.notFound();
        }
    }
    else {                                                              <i class="conum" data-value="3"></i><b>(3)</b>
        LOG.info(
            "Found data for key {} on remote host {}:{}",
            id,
            metadata.host(),
            metadata.port()
        );
        return GetWeatherStationDataResult.foundRemotely(metadata.host(), metadata.port());
    }
}

public List&lt;PipelineMetadata&gt; getMetaData() {                           <i class="conum" data-value="4"></i><b>(4)</b>
    return streams.allMetadataForStore(TopologyProducer.WEATHER_STATIONS_STORE)
            .stream()
            .map(m -&gt; new PipelineMetadata(
                    m.hostInfo().host() + ":" + m.hostInfo().port(),
                    m.topicPartitions()
                        .stream()
                        .map(TopicPartition::toString)
                        .collect(Collectors.toSet()))
            )
            .collect(Collectors.toList());
}</code></pre>
</div>
</div>
<div class="colist arabic">
<table>
<tr>
<td><i class="conum" data-value="1"></i><b>1</b></td>
<td>The streams metadata for the given weather station id is obtained</td>
</tr>
<tr>
<td><i class="conum" data-value="2"></i><b>2</b></td>
<td>The given key (weather station id) is maintained by the local application node, i.e. it can answer the query itself</td>
</tr>
<tr>
<td><i class="conum" data-value="3"></i><b>3</b></td>
<td>The given key is maintained by another application node; in this case the information about that node (host and port) will be returned</td>
</tr>
<tr>
<td><i class="conum" data-value="4"></i><b>4</b></td>
<td>The <code>getMetaData()</code> method is added to provide callers with a list of all the nodes in the application cluster.</td>
</tr>
</table>
</div>
<div class="paragraph">
<p>The <code>GetWeatherStationDataResult</code> type must be adjusted accordingly:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="java" class="language-java hljs">package org.acme.kafka.streams.aggregator.streams;

import java.util.Optional;
import java.util.OptionalInt;

import org.acme.kafka.streams.aggregator.model.WeatherStationData;

public class GetWeatherStationDataResult {

    private static GetWeatherStationDataResult NOT_FOUND =
            new GetWeatherStationDataResult(null, null, null);

    private final WeatherStationData result;
    private final String host;
    private final Integer port;

    private GetWeatherStationDataResult(WeatherStationData result, String host,
            Integer port) {
        this.result = result;
        this.host = host;
        this.port = port;
    }

    public static GetWeatherStationDataResult found(WeatherStationData data) {
        return new GetWeatherStationDataResult(data, null, null);
    }

    public static GetWeatherStationDataResult foundRemotely(String host, int port) {
        return new GetWeatherStationDataResult(null, host, port);
    }

    public static GetWeatherStationDataResult notFound() {
        return NOT_FOUND;
    }

    public Optional&lt;WeatherStationData&gt; getResult() {
        return Optional.ofNullable(result);
    }

    public Optional&lt;String&gt; getHost() {
        return Optional.ofNullable(host);
    }

    public OptionalInt getPort() {
        return port != null ? OptionalInt.of(port) : OptionalInt.empty();
    }
}</code></pre>
</div>
</div>
<div class="paragraph">
<p>Also the return type for <code>getMetaData()</code> must be defined
(<code>aggregator/src/main/java/org/acme/kafka/streams/aggregator/streams/PipelineMetadata.java</code>):</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="java" class="language-java hljs">package org.acme.kafka.streams.aggregator.streams;

import java.util.Set;

public class PipelineMetadata {

    public String host;
    public Set&lt;String&gt; partitions;

    public PipelineMetadata(String host, Set&lt;String&gt; partitions) {
        this.host = host;
        this.partitions = partitions;
    }
}</code></pre>
</div>
</div>
<div class="paragraph">
<p>Lastly, the REST endpoint class must be updated:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="java" class="language-java hljs">package org.acme.kafka.streams.aggregator.rest;

import java.net.URI;
import java.net.URISyntaxException;
import java.util.List;

import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
import javax.ws.rs.Consumes;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;

import org.acme.kafka.streams.aggregator.streams.GetWeatherStationDataResult;
import org.acme.kafka.streams.aggregator.streams.KafkaStreamsPipeline;
import org.acme.kafka.streams.aggregator.streams.PipelineMetadata;

@ApplicationScoped
@Path("/weather-stations")
public class WeatherStationEndpoint {

    @Inject
    InteractiveQueries interactiveQueries;

    @GET
    @Path("/data/{id}")
    @Consumes(MediaType.APPLICATION_JSON)
    @Produces(MediaType.APPLICATION_JSON)
    public Response getWeatherStationData(@PathParam("id") int id) {
        GetWeatherStationDataResult result = interactiveQueries.getWeatherStationData(id);

        if (result.getResult().isPresent()) {                     <i class="conum" data-value="1"></i><b>(1)</b>
            return Response.ok(result.getResult().get()).build();
        }
        else if (result.getHost().isPresent()) {                  <i class="conum" data-value="2"></i><b>(2)</b>
            URI otherUri = getOtherUri(result.getHost().get(), result.getPort().getAsInt(),
                    id);
            return Response.seeOther(otherUri).build();
        }
        else {                                                    <i class="conum" data-value="3"></i><b>(3)</b>
            return Response.status(Status.NOT_FOUND.getStatusCode(),
                    "No data found for weather station " + id).build();
        }
    }

    @GET
    @Path("/meta-data")
    @Produces(MediaType.APPLICATION_JSON)
    public List&lt;PipelineMetadata&gt; getMetaData() {                 <i class="conum" data-value="4"></i><b>(4)</b>
        return interactiveQueries.getMetaData();
    }

    private URI getOtherUri(String host, int port, int id) {
        try {
            return new URI("http://" + host + ":" + port + "/weather-stations/data/" + id);
        }
        catch (URISyntaxException e) {
            throw new RuntimeException(e);
        }
    }
}</code></pre>
</div>
</div>
<div class="colist arabic">
<table>
<tr>
<td><i class="conum" data-value="1"></i><b>1</b></td>
<td>The data was found locally, so return it</td>
</tr>
<tr>
<td><i class="conum" data-value="2"></i><b>2</b></td>
<td>The data is maintained by another node, so reply with a redirect (HTTP status code 303) if the data for the given key is stored on one of the other nodes.</td>
</tr>
<tr>
<td><i class="conum" data-value="3"></i><b>3</b></td>
<td>No data was found for the given weather station id</td>
</tr>
<tr>
<td><i class="conum" data-value="4"></i><b>4</b></td>
<td>Exposes information about all the hosts forming the application cluster</td>
</tr>
</table>
</div>
<div class="paragraph">
<p>Now stop the <code>aggregator</code> service again and rebuild it.
Then let&#8217;s spin up three instances of it:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs">./mvnw clean package -f aggregator/pom.xml
docker-compose stop aggregator
docker-compose up --build -d --scale aggregator=3</code></pre>
</div>
</div>
<div class="paragraph">
<p>When invoking the REST API on any of the three instances, it might either be
that the aggregation for the requested weather station id is stored locally on the node receiving the query,
or it could be stored on one of the other two nodes.</p>
</div>
<div class="paragraph">
<p>As the load balancer of Docker Compose will distribute requests to the <code>aggregator</code> service in a round-robin fashion,
we&#8217;ll invoke the actual nodes directly.
The application exposes information about all the host names via REST:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs">http aggregator:8080/weather-stations/meta-data

HTTP/1.1 200 OK
Connection: keep-alive
Content-Length: 202
Content-Type: application/json
Date: Tue, 18 Jun 2019 20:00:23 GMT

[
    {
        "host": "2af13fe516a9:8080",
        "partitions": [
            "temperature-values-2"
        ]
    },
    {
        "host": "32cc8309611b:8080",
        "partitions": [
            "temperature-values-1"
        ]
    },
    {
        "host": "1eb39af8d587:8080",
        "partitions": [
            "temperature-values-0"
        ]
    }
]</code></pre>
</div>
</div>
<div class="paragraph">
<p>Retrieve the data from one of the three hosts shown in the response
(your actual host names will differ):</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs">http 2af13fe516a9:8080/weather-stations/data/1</code></pre>
</div>
</div>
<div class="paragraph">
<p>If that node holds the data for key "1", you&#8217;ll get a response like this:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs">HTTP/1.1 200 OK
Connection: keep-alive
Content-Length: 74
Content-Type: application/json
Date: Tue, 11 Jun 2019 19:16:31 GMT

{
  "avg": 11.9,
  "count": 259,
  "max": 50.0,
  "min": -30.1,
  "stationId": 1,
  "stationName": "Hamburg"
}</code></pre>
</div>
</div>
<div class="paragraph">
<p>Otherwise, the service will send a redirect:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs">HTTP/1.1 303 See Other
Connection: keep-alive
Content-Length: 0
Date: Tue, 18 Jun 2019 20:01:03 GMT
Location: http://1eb39af8d587:8080/weather-stations/data/1</code></pre>
</div>
</div>
<div class="paragraph">
<p>You can also have <em>httpie</em> automatically follow the redirect by passing the <code>--follow option</code>:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="bash" class="language-bash hljs">http --follow 2af13fe516a9:8080/weather-stations/data/1</code></pre>
</div>
</div>
</div>
</div>
<div class="sect1">
<h2 id="running-natively"><a class="anchor" href="#running-natively"></a>Running Natively</h2>
<div class="sectionbody">
<div class="paragraph">
<p>The Quarkus extension for Kafka Streams enables the execution of stream processing applications
natively via GraalVM without further configuration.</p>
</div>
<div class="paragraph">
<p>To run both the <code>producer</code> and <code>aggregator</code> applications in native mode,
the Maven builds can be executed using the <code>native</code> profile:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="shell" class="language-shell hljs">./mvnw clean package -f producer/pom.xml -Pnative -Dnative-image.container-runtime=docker
./mvnw clean package -f aggregator/pom.xml -Pnative -Dnative-image.container-runtime=docker</code></pre>
</div>
</div>
<div class="paragraph">
<p>Now create an environment variable named <code>QUARKUS_MODE</code> and with value set to "native":</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="shell" class="language-shell hljs">export QUARKUS_MODE=native</code></pre>
</div>
</div>
<div class="paragraph">
<p>This is used by the Docker Compose file to use the correct <code>Dockerfile</code> when building the <code>producer</code> and <code>aggregator</code> images.
The Kafka Streams application can work with less than 50 MB RSS in native mode.
To do so, add the <code>Xmx</code> option to the program invocation in <code>aggregator/src/main/docker/Dockerfile.native</code>:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code data-lang="shell" class="language-shell hljs">CMD ["./application", "-Dquarkus.http.host=0.0.0.0", "-Xmx32m"]</code></pre>
</div>
</div>
<div class="paragraph">
<p>Now start Docker Compose as described above
(don&#8217;t forget to rebuild the container images).</p>
</div>
</div>
</div>
<div class="sect1">
<h2 id="kafka-streams-health-checks"><a class="anchor" href="#kafka-streams-health-checks"></a>Kafka Streams Health Checks</h2>
<div class="sectionbody">
<div class="paragraph">
<p>If you are using the <code>quarkus-smallrye-health</code> extension, <code>quarkus-kafka-streams</code> will automatically add:</p>
</div>
<div class="ulist">
<ul>
<li>
<p>a readiness health check to validate that all topics declared in the <code>quarkus.kafka-streams.topics</code> property are created,</p>
</li>
<li>
<p>a liveness health check based on the Kafka Streams state.</p>
</li>
</ul>
</div>
<div class="paragraph">
<p>So when you access the <code>/health</code> endpoint of your application you will have information about the state of the Kafka Streams and the available and/or missing topics.</p>
</div>
<div class="paragraph">
<p>This is an example of when the status is <code>DOWN</code>:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs">curl -i http://aggregator:8080/health

HTTP/1.1 503 Service Unavailable
content-type: application/json; charset=UTF-8
content-length: 454

{
    "status": "DOWN",
    "checks": [
        {
            "name": "Kafka Streams state health check",  <i class="conum" data-value="1"></i><b>(1)</b>
            "status": "DOWN",
            "data": {
                "state": "CREATED"
            }
        },
        {
            "name": "Kafka Streams topics health check",  <i class="conum" data-value="2"></i><b>(2)</b>
            "status": "DOWN",
            "data": {
                "available_topics": "weather-stations,temperature-values",
                "missing_topics": "hygrometry-values"
            }
        }
    ]
}</code></pre>
</div>
</div>
<div class="colist arabic">
<table>
<tr>
<td><i class="conum" data-value="1"></i><b>1</b></td>
<td>Liveness health check. Also available at <code>/health/live</code> endpoint.</td>
</tr>
<tr>
<td><i class="conum" data-value="2"></i><b>2</b></td>
<td>Readiness health check. Also available at <code>/health/ready</code> endpoint.</td>
</tr>
</table>
</div>
<div class="paragraph">
<p>So as you can see, the status is <code>DOWN</code> as soon as one of the <code>quarkus.kafka-streams.topics</code> is missing or the Kafka Streams <code>state</code> is not <code>RUNNING</code>.</p>
</div>
<div class="paragraph">
<p>If no topics are available, the <code>available_topics</code> key will not be present in the <code>data</code> field of the <code>Kafka Streams topics health check</code>.
As well as if no topics are missing, the <code>missing_topics</code> key will not be present in the <code>data</code> field of the <code>Kafka Streams topics health check</code>.</p>
</div>
<div class="paragraph">
<p>You can of course disable the health check of the <code>quarkus-kafka-streams</code> extension by setting the <code>quarkus.kafka-streams.health.enabled</code> property to <code>false</code> in your <code>application.properties</code>.</p>
</div>
<div class="paragraph">
<p>Obviously you can create your liveness and readiness probes based on the respective endpoints <code>/health/live</code> and <code>/health/ready</code>.</p>
</div>
<div class="sect2">
<h3 id="liveness-health-check"><a class="anchor" href="#liveness-health-check"></a>Liveness health check</h3>
<div class="paragraph">
<p>Here is an example of the liveness check:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs">curl -i http://aggregator:8080/health/live

HTTP/1.1 503 Service Unavailable
content-type: application/json; charset=UTF-8
content-length: 225

{
    "status": "DOWN",
    "checks": [
        {
            "name": "Kafka Streams state health check",
            "status": "DOWN",
            "data": {
                "state": "CREATED"
            }
        }
    ]
}</code></pre>
</div>
</div>
<div class="paragraph">
<p>The <code>state</code> is coming from the <code>KafkaStreams.State</code> enum.</p>
</div>
</div>
<div class="sect2">
<h3 id="readiness-health-check"><a class="anchor" href="#readiness-health-check"></a>Readiness health check</h3>
<div class="paragraph">
<p>Here is an example of the readiness check:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-none hljs">curl -i http://aggregator:8080/health/ready

HTTP/1.1 503 Service Unavailable
content-type: application/json; charset=UTF-8
content-length: 265

{
    "status": "DOWN",
    "checks": [
        {
            "name": "Kafka Streams topics health check",
            "status": "DOWN",
            "data": {
                "missing_topics": "weather-stations,temperature-values"
            }
        }
    ]
}</code></pre>
</div>
</div>
</div>
</div>
</div>
<div class="sect1">
<h2 id="going-further"><a class="anchor" href="#going-further"></a>Going Further</h2>
<div class="sectionbody">
<div class="paragraph">
<p>This guide has shown how you can build stream processing applications using Quarkus and the Kafka Streams APIs,
both in JVM and native modes.
For running your KStreams application in production, you could also add health checks and metrics for the data pipeline.
Refer to the Quarkus guides on <a href="microprofile-metrics">metrics</a> and <a href="microprofile-health">health checks</a> to learn more.</p>
</div>
</div>
</div>
<div class="sect1">
<h2 id="configuration-reference"><a class="anchor" href="#configuration-reference"></a>Configuration Reference</h2>
<div class="sectionbody">
<div class="paragraph configuration-legend">
<p><span class="icon"><i class="fa fa-lock" title="Fixed at build time"></i></span> Configuration property fixed at build time - All other configuration properties are overridable at runtime</p>
</div>
<table class="tableblock frame-all grid-all stretch configuration-reference searchable">
<colgroup>
<col style="width: 80%;">
<col style="width: 10%;">
<col style="width: 10%;">
</colgroup>
<tbody>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><a id="quarkus-kafka-streams_configuration"></a><a href="#quarkus-kafka-streams_configuration">Configuration property</a></p></th>
<th class="tableblock halign-left valign-middle"><p class="tableblock">Type</p></th>
<th class="tableblock halign-left valign-middle"><p class="tableblock">Default</p></th>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><span class="icon"><i class="fa fa-lock" title="Fixed at build time"></i></span> <a id="quarkus-kafka-streams_quarkus.kafka-streams.health.enabled"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.health.enabled">quarkus.kafka-streams.health.enabled</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>Whether or not a health check is published in case the smallrye-health extension is present (defaults to true).</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">boolean</p></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock"><code>true</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.application-id"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.application-id">quarkus.kafka-streams.application-id</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>A unique identifier for this Kafka Streams application. If not set, defaults to quarkus.application.name.</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock"><code>${quarkus.application.name}</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.bootstrap-servers"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.bootstrap-servers">quarkus.kafka-streams.bootstrap-servers</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>A comma-separated list of host:port pairs identifying the Kafka bootstrap server(s)</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">list of host:port</p></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock"><code>localhost:9012</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.application-server"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.application-server">quarkus.kafka-streams.application-server</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>A unique identifier of this application instance, typically in the form host:port.</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.topics"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.topics">quarkus.kafka-streams.topics</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>A comma-separated list of topic names. The pipeline will only be started once all these topics are present in the Kafka cluster.</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">list of string</p></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">required <span class="icon"><i class="fa fa-exclamation-circle" title="Configuration property is required"></i></span></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.schema-registry-key"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.schema-registry-key">quarkus.kafka-streams.schema-registry-key</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>The schema registry key. e.g. to diff between different registry impls / instances as they have this registry url under different property key. Red Hat / Apicurio - apicurio.registry.url Confluent - schema.registry.url</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock"><code>schema.registry.url</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.schema-registry-url"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.schema-registry-url">quarkus.kafka-streams.schema-registry-url</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>The schema registry url.</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.security.protocol"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.security.protocol">quarkus.kafka-streams.security.protocol</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>The security protocol to use See <a href="https://docs.confluent.io/current/streams/developer-guide/security.html#security-example" class="bare">https://docs.confluent.io/current/streams/developer-guide/security.html#security-example</a></p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.sasl.jaas-config"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.sasl.jaas-config">quarkus.kafka-streams.sasl.jaas-config</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>JAAS login context parameters for SASL connections in the format used by JAAS configuration files</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.sasl.client-callback-handler-class"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.sasl.client-callback-handler-class">quarkus.kafka-streams.sasl.client-callback-handler-class</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>The fully qualified name of a SASL client callback handler class</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.sasl.login-callback-handler-class"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.sasl.login-callback-handler-class">quarkus.kafka-streams.sasl.login-callback-handler-class</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>The fully qualified name of a SASL login callback handler class</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.sasl.login-class"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.sasl.login-class">quarkus.kafka-streams.sasl.login-class</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>The fully qualified name of a class that implements the Login interface</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.sasl.kerberos-service-name"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.sasl.kerberos-service-name">quarkus.kafka-streams.sasl.kerberos-service-name</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>The Kerberos principal name that Kafka runs as</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.sasl.kerberos-kinit-cmd"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.sasl.kerberos-kinit-cmd">quarkus.kafka-streams.sasl.kerberos-kinit-cmd</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>Kerberos kinit command path</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.sasl.kerberos-ticket-renew-window-factor"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.sasl.kerberos-ticket-renew-window-factor">quarkus.kafka-streams.sasl.kerberos-ticket-renew-window-factor</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>Login thread will sleep until the specified window factor of time from last refresh</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">double</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.sasl.kerberos-ticket-renew-jitter"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.sasl.kerberos-ticket-renew-jitter">quarkus.kafka-streams.sasl.kerberos-ticket-renew-jitter</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>Percentage of random jitter added to the renewal time</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">double</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.sasl.kerberos-min-time-before-relogin"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.sasl.kerberos-min-time-before-relogin">quarkus.kafka-streams.sasl.kerberos-min-time-before-relogin</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>Percentage of random jitter added to the renewal time</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">long</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.sasl.login-refresh-window-factor"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.sasl.login-refresh-window-factor">quarkus.kafka-streams.sasl.login-refresh-window-factor</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>Login refresh thread will sleep until the specified window factor relative to the credential&#8217;s lifetime has been reached-</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">double</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.sasl.login-refresh-window-jitter"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.sasl.login-refresh-window-jitter">quarkus.kafka-streams.sasl.login-refresh-window-jitter</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>The maximum amount of random jitter relative to the credential&#8217;s lifetime</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">double</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.sasl.login-refresh-min-period"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.sasl.login-refresh-min-period">quarkus.kafka-streams.sasl.login-refresh-min-period</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>The desired minimum duration for the login refresh thread to wait before refreshing a credential</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock"><a href="https://docs.oracle.com/javase/8/docs/api/java/time/Duration.html">Duration</a>
  <a href="#duration-note-anchor" title="More information about the Duration format"><span class="icon"><i class="fa fa-question-circle"></i></span></a></p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.sasl.login-refresh-buffer"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.sasl.login-refresh-buffer">quarkus.kafka-streams.sasl.login-refresh-buffer</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>The amount of buffer duration before credential expiration to maintain when refreshing a credential</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock"><a href="https://docs.oracle.com/javase/8/docs/api/java/time/Duration.html">Duration</a>
  <a href="#duration-note-anchor" title="More information about the Duration format"><span class="icon"><i class="fa fa-question-circle"></i></span></a></p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.ssl.protocol"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.ssl.protocol">quarkus.kafka-streams.ssl.protocol</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>The SSL protocol used to generate the SSLContext</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.ssl.provider"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.ssl.provider">quarkus.kafka-streams.ssl.provider</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>The name of the security provider used for SSL connections</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.ssl.cipher-suites"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.ssl.cipher-suites">quarkus.kafka-streams.ssl.cipher-suites</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>A list of cipher suites</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.ssl.enabled-protocols"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.ssl.enabled-protocols">quarkus.kafka-streams.ssl.enabled-protocols</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>The list of protocols enabled for SSL connections</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.ssl.truststore.type"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.ssl.truststore.type">quarkus.kafka-streams.ssl.truststore.type</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>Store type</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.ssl.truststore.location"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.ssl.truststore.location">quarkus.kafka-streams.ssl.truststore.location</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>Store location</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.ssl.truststore.password"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.ssl.truststore.password">quarkus.kafka-streams.ssl.truststore.password</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>Store password</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.ssl.keystore.type"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.ssl.keystore.type">quarkus.kafka-streams.ssl.keystore.type</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>Store type</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.ssl.keystore.location"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.ssl.keystore.location">quarkus.kafka-streams.ssl.keystore.location</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>Store location</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.ssl.keystore.password"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.ssl.keystore.password">quarkus.kafka-streams.ssl.keystore.password</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>Store password</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.ssl.key.type"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.ssl.key.type">quarkus.kafka-streams.ssl.key.type</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>Store type</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.ssl.key.location"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.ssl.key.location">quarkus.kafka-streams.ssl.key.location</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>Store location</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.ssl.key.password"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.ssl.key.password">quarkus.kafka-streams.ssl.key.password</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>Store password</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.ssl.keymanager-algorithm"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.ssl.keymanager-algorithm">quarkus.kafka-streams.ssl.keymanager-algorithm</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>The algorithm used by key manager factory for SSL connections</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.ssl.trustmanager-algorithm"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.ssl.trustmanager-algorithm">quarkus.kafka-streams.ssl.trustmanager-algorithm</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>The algorithm used by trust manager factory for SSL connections</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.ssl.endpoint-identification-algorithm"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.ssl.endpoint-identification-algorithm">quarkus.kafka-streams.ssl.endpoint-identification-algorithm</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>The endpoint identification algorithm to validate server hostname using server certificate</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock"><code>https</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p><a id="quarkus-kafka-streams_quarkus.kafka-streams.ssl.secure-random-implementation"></a><code><a href="#quarkus-kafka-streams_quarkus.kafka-streams.ssl.secure-random-implementation">quarkus.kafka-streams.ssl.secure-random-implementation</a></code></p>
</div>
<div class="openblock description">
<div class="content">
<div class="paragraph">
<p>The SecureRandom PRNG implementation to use for SSL cryptography operations</p>
</div>
</div>
</div></div></td>
<td class="tableblock halign-left valign-middle"><p class="tableblock">string</p></td>
<td class="tableblock halign-left valign-middle"></td>
</tr>
</tbody>
</table>
<div id="duration-note-anchor" class="admonitionblock note">
<table>
<tr>
<td class="icon">
<i class="fa icon-note" title="Note"></i>
</td>
<td class="content">
<div class="title">About the Duration format</div>
<div class="paragraph">
<p>The format for durations uses the standard <code>java.time.Duration</code> format.
You can learn more about it in the <a href="https://docs.oracle.com/javase/8/docs/api/java/time/Duration.html#parse-java.lang.CharSequence-">Duration#parse() javadoc</a>.</p>
</div>
<div class="paragraph">
<p>You can also provide duration values starting with a number.
In this case, if the value consists only of a number, the converter treats the value as seconds.
Otherwise, <code>PT</code> is implicitly prepended to the value to obtain a standard <code>java.time.Duration</code> format.</p>
</div>
</td>
</tr>
</table>
</div>
</div>
</div>
    </div>
  </div>
</div>

  </div>

  <div class="content project-footer">
  <div class="footer-section">
    <div class="logo-wrapper">
      <a href="/"><img src="/assets/images/quarkus_logo_horizontal_rgb_reverse.svg" class="project-logo" title="Quarkus"></a>
    </div>
  </div>
  <div class="grid-wrapper">
    <p class="grid__item width-3-12">Quarkus is open. All dependencies of this project are available under the <a href='https://www.apache.org/licenses/LICENSE-2.0' target='_blank'>Apache Software License 2.0</a> or compatible license.<br /><br />This website was built with <a href='https://jekyllrb.com/' target='_blank'>Jekyll</a>, is hosted on <a href='https://pages.github.com/' target='_blank'>Github Pages</a> and is completely open source. If you want to make it better, <a href='https://github.com/quarkusio/quarkusio.github.io' target='_blank'>fork the website</a> and show us what you’ve got.</p>

    
      <div class="width-1-12 project-links">
        <span>Navigation</span>
        <ul class="footer-links width-1-12">
          
            <li><a href="/">Home</a></li>
          
            <li><a href="/guides">Guides</a></li>
          
            <li><a href="/community/#contributing">Contribute</a></li>
          
            <li><a href="/faq">FAQ</a></li>
          
            <li><a href="/get-started">Get Started</a></li>
          
        </ul>
      </div>
    
      <div class="width-1-12 project-links">
        <span>Contribute</span>
        <ul class="footer-links width-1-12">
          
            <li><a href="https://twitter.com/quarkusio">Follow us</a></li>
          
            <li><a href="https://github.com/quarkusio">GitHub</a></li>
          
            <li><a href="/security">Security&nbsp;policy</a></li>
          
        </ul>
      </div>
    
      <div class="width-1-12 project-links">
        <span>Get Help</span>
        <ul class="footer-links width-1-12">
          
            <li><a href="https://groups.google.com/forum/#!forum/quarkus-dev">Forums</a></li>
          
            <li><a href="https://quarkusio.zulipchat.com">Chatroom</a></li>
          
        </ul>
      </div>
    

    
      <div class="width-3-12 more-links">
        <span>Quarkus is made of community projects</span>
        <ul class="footer-links">
          
            <li><a href="https://vertx.io/" target="_blank">Eclipse Vert.x</a></li>
          
            <li><a href="https://microprofile.io" target="_blank">Eclipse MicroProfile</a></li>
          
            <li><a href="https://hibernate.org" target="_blank">Hibernate</a></li>
          
            <li><a href="https://netty.io" target="_blank">Netty</a></li>
          
            <li><a href="https://resteasy.github.io" target="_blank">RESTEasy</a></li>
          
            <li><a href="https://camel.apache.org" target="_blank">Apache Camel</a></li>
          
            <li><a href="https://code.quarkus.io/" target="_blank">And many more...</a></li>
          
        </ul>
      </div>
    
  </div>
</div>
  <div class="content redhat-footer">
  <div class="grid-wrapper">
    <span class="licence">
      <i class="fab fa-creative-commons"></i><i class="fab fa-creative-commons-by"></i> <a href="https://creativecommons.org/licenses/by/3.0/" target="_blank">CC by 3.0</a> | <a href="https://www.redhat.com/en/about/privacy-policy">Privacy Policy</a>
    </span>
    <span class="redhat">
      Sponsored by
    </span>
    <span class="redhat-logo">
      <a href="https://www.redhat.com/" target="_blank"><img src="/assets/images/redhat_reversed.svg"></a>
    </span>
  </div>
</div>


  <script src="https://ajax.googleapis.com/ajax/libs/jquery/2.1.4/jquery.min.js" integrity="sha384-8gBf6Y4YYq7Jx97PIqmTwLPin4hxIzQw5aDmUg/DDhul9fFpbbLcLh3nTIIDJKhx" crossorigin="anonymous"></script>
  <script type="text/javascript" src="/assets/javascript/mobile-nav.js"></script>
  <script type="text/javascript" src="/assets/javascript/scroll-down.js"></script>
  <script src="/assets/javascript/satellite.js" type="text/javascript"></script>
  <script src="https://quarkus.io/guides/javascript/config.js" type="text/javascript"></script>
  <script src="/assets/javascript/search-filter.js" type="text/javascript"></script>
  <script src="/assets/javascript/back-to-top.js" type="text/javascript"></script>
</body>

</html>
